package com.storm.lesson.transaction1;

import java.util.HashMap;
import java.util.Map;
import java.util.Random;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.transactional.ITransactionalSpout;
import backtype.storm.tuple.Fields;

/**
 * 事务性的spout需要实现ITransactionalSpout，这个接口包含两个内部接口类Coordinator和Emitter。
 * 在topology运行的时候，事务性的spout内部包含一个子Topology
 * coordinator只有一个，emitter根据并行度可以有多个实例
 * * * 1.这里面有两种类型的tuple，一种是事务性的tuple，一种是batch中的tuple；
 * * * 2.coordinator开启一个事务准备发射一个batch时候，进入一个事务的processing阶段，
 * * * 会发射一个事务性tuple(transactionAttempt & metadata)到"batch emit"流
 * * * 3.coordinator的"batch emit"流，负责为每个batch实际发射tuple。
 * * * 发送的tuple都必须以TransactionAttempt作为第一个field，storm根据这个field来判断tuple属于哪一个batch。
 * * * 4.Emitter以all grouping(广播)的方式订阅
 * @author zhihu
 *
 */
public class MyTxSpout implements ITransactionalSpout {

	private static final long serialVersionUID = 1L;

	/**
	 * 数据源
	 */
	Map<Long, String> dbMap = null;

	public MyTxSpout() {
		Random random = new Random();
		dbMap = new HashMap<Long, String>();
		String[] hosts = { "www.taobao.com" };
		String[] session_id = { "ABYH6Y4V4SCVXTG6DPB4VH9U123",
				"XXYH6YCGFJYERTT834R52FDXV9U34",
				"BBYH61456FGHHJ7JL89RG5VV9UYU7", "CYYH6Y2345GHI899OFG4V9U567",
				"VVVYH6Y4V4SFXZ56JIPDPB4V678" };
		String[] time = { "2014-01-07 08:40:50", "2014-01-07 08:40:51",
				"2014-01-07 08:40:52", "2014-01-07 08:40:53",
				"2014-01-07 09:40:49", "2014-01-07 10:40:49",
				"2014-01-07 11:40:49", "2014-01-07 12:40:49" };

		for (long i = 0; i < 100; i++) {
			dbMap.put(i, hosts[0] + "\t" + session_id[random.nextInt(5)] + "\t"
					+ time[random.nextInt(8)]);
		}
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("tx", "log"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		return null;
	}

	/**
	 * coordinator开启一个事务准备发射一个batch时候，进入一个事务的processing阶段，会发射一个事务性tuple(transactionAttempt & metadata)到"batch emit"流
	 * coordinator的"batch emit"流，负责为每个batch实际发射tuple。发送的tuple都必须以TransactionAttempt作为第一个field，storm根据这个field来判断tuple属于哪一个batch。
	 */
	@Override
	public Coordinator getCoordinator(Map conf, TopologyContext context) {
		return new MyCoordinator();
	}

	@Override
	public Emitter getEmitter(Map conf, TopologyContext context) {
		return new MyEmitter(dbMap);
	}

}
